package com.doit.doitdata.process

import java.util
import java.util.ArrayList

import ch.hsr.geohash.GeoHash
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.doit.doitdata.bean.LogBean
import org.apache.commons.httpclient.HttpClient
import org.apache.commons.httpclient.methods.GetMethod
import org.apache.commons.lang.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import redis.clients.jedis.Jedis

/**
  * Created by hunter.coder 涛哥  
  * 2019/3/28 17:42
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  * Description:
  * 性能测试：  5000条日志，耗时868s ==> 14.5分钟
  * 可知，如果纯粹靠请求高德地图服务来做gps地理位置解析，是基本上行不通的
  **/
object Preprocess {

  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.ERROR)
    val inpath = "G:\\logs\\2019-04-17"
    val outpath = "G:\\logs\\2019-04-17-out"

    val spark = SparkSession.builder()
      .enableHiveSupport()
      .appName(Preprocess.getClass.getSimpleName)
      .master("local")
      .getOrCreate()
    import spark.implicits._

    val start = System.currentTimeMillis()
    // 读取原始日志
    val rawDs = spark.read.textFile(inpath).filter(line => line.split(" --> ").size > 1)
    val logBeanRDD: RDD[LogBean] = rawDs.rdd.mapPartitions(part => {

      // 构建一些公用工具，比如jedis客户端
      val jedis = new Jedis("c701", 6379)
      // 构造一个http客户端对象
      val httpClient: HttpClient = new HttpClient

      // 再对分区中的每一条记录进行map变换
      val res: Iterator[LogBean] = part.map(line => {
        val json = line.split(" --> ")(1)
        var jsonObj = new JSONObject()
        try {
          jsonObj = JSON.parseObject(json)
        } catch {
          case e: Exception => {e.printStackTrace()}
        }

        val uObj = jsonObj.getJSONObject("u")
        val phoneObj = uObj.getJSONObject("phone")
        val locObj = uObj.getJSONObject("loc")
        val appObj = uObj.getJSONObject("app")

        // 取出user对象中的扁平字段
        val id = uObj.getString("id")
        val account = uObj.getString("account")
        val sessionId = uObj.getString("sessionId")

        // 取出手机设备信息
        val imei = phoneObj.getString("imei")
        val osName = phoneObj.getString("osName")
        val osVer = phoneObj.getString("osVer")
        val resolution = phoneObj.getString("resolution")
        val androidId = phoneObj.getString("androidId")
        val manufacture = phoneObj.getString("manufacture")
        val deviceId = phoneObj.getString("deviceId")

        // 取出loc位置信息
        val areacode = locObj.getString("areacode")
        val longtitude = locObj.getDouble("longtitude")
        val latitude = locObj.getDouble("latitude")
        val carrier = locObj.getString("carrier")
        val netType = locObj.getString("netType")
        val cid_sn = locObj.getString("cid_sn")
        val ip = locObj.getString("ip")

        // 取出app各个字段
        val appid = appObj.getString("appid")
        val appVer = appObj.getString("appVer")
        val release_ch = appObj.getString("release_ch")
        val promotion_ch = appObj.getString("promotion_ch")


        val logType = jsonObj.getString("logType")
        val commit_time = jsonObj.getLong("commit_time")

        val eventObj = jsonObj.getJSONObject("event")
        // 构造一个用于装event数据的hashmap
        val eventMap = new scala.collection.mutable.HashMap[String, String]()
        val iter = eventObj.keySet().iterator()
        // 迭代取出event中每一对kv
        while (iter.hasNext) {
          val k = iter.next()
          val v = eventObj.getString(k)
          // 添加到hashmap中
          eventMap.put(k, v)
        }


        // 将经纬度解析成地理位置信息（省市区、商圈）
        var province: String = ""
        var city: String = ""
        var district: String = ""
        var bizStr: String = ""

        // 1.先去公司自己的redis中查找位置信息
        val geohashCode = GeoHash.withCharacterPrecision(latitude, longtitude, 8).toBase32
        // 湖南省,长沙市,岳麓区|岳麓书院,橘子洲
        val str = jedis.get(geohashCode)
        if (null != str) {

          val pair = str.split("\\|")

          // 获取省市区
          val pcd = pair(0).split(",")

          // 获取商圈
          val bizs = if (pair.length > 1) pair(1).split(",") else Array("")

          province = pcd(0)
          city = pcd(1)
          district = pcd(2)

          bizStr = bizs.mkString("", " ", "")

        } else if (1 == 0) {
          // 2.如果查找不到，再去高德请求服务


          // 构造一个get请求对象
          //val key = "e384b6b8bc2f8e9e9e92a9cf969da45c"
          //val key = "41e7ce0e00fc0d5eaeaccb4cc1f30297"
          val key = "4a798bd4d57d9a7bf52f936079d82d16"
          val get: GetMethod = new GetMethod("https://restapi.amap.com/v3/geocode/regeo?key=" + key + "&location=" + longtitude + "," + latitude)

          // 用客户端去执行构造好的get请求，请求的结果会直接封装到get对象中
          httpClient.executeMethod(get)

          //从get对象中获取响应
          val resJson: String = get.getResponseBodyAsString

          // 获取省市区及商圈信息
          val jsonObject: JSONObject = JSON.parseObject(resJson)

          val regeocode: JSONObject = jsonObject.getJSONObject("regeocode")
          val addressComponent: JSONObject = regeocode.getJSONObject("addressComponent")

          province = addressComponent.getString("province")
          city = addressComponent.getString("city")
          district = addressComponent.getString("district")

          // 获取商圈信息列表
          val businessAreas: JSONArray = addressComponent.getJSONArray("businessAreas")

          // val bizList: ArrayList[JSONObject] = new ArrayList[JSONObject]

          val sb = new StringBuilder

          // 遍历返回的所有商圈
          for (i <- 0 until businessAreas.size() if businessAreas.size() > 0) {

            val biz = try {
              businessAreas.getJSONObject(i)
            } catch {
              case e: Exception => null
            }

            if (null != biz) {
              // 获取商圈名称
              val bizName = biz.getString("name")

              // 获取坐标点
              val gpsStr = biz.getString("location").split(",")

              // 将商圈的中心点坐标 变换成  geohash码
              val geoBiz = GeoHash.withCharacterPrecision(gpsStr(1).toDouble, gpsStr(0).toDouble, 8).toBase32

              // 将该商圈的geohash码及 地理位置信息（省市区，商圈）存入公司自己的知识库
              jedis.set(geoBiz, province + "," + city + "," + district + "|" + bizName)


              // 把每一个商圈名称拼接到一个统一的字符串中
              sb.append(bizName)
            }

          }


          // 拼接好的商圈名称结果
          bizStr = sb.mkString("", " ", "")

          // 将查询到的结果写入公司自己的地理位置信息知识库
          jedis.set(geohashCode, province + "," + city + "," + district + "|" + bizStr)
        }

        // 组装数据并返回
        LogBean(id,
          account,
          sessionId,
          imei,
          osName,
          osVer,
          resolution,
          androidId,
          manufacture,
          deviceId,
          areacode,
          longtitude,
          latitude,
          carrier,
          netType,
          cid_sn,
          ip,
          appid,
          appVer,
          release_ch,
          promotion_ch,
          logType,
          commit_time,
          province,
          city,
          district,
          bizStr,
          eventMap)
      })

      jedis.close()

      res
    })

      .filter(bean => StringUtils.isNotBlank(bean.id + bean.account + bean.imei + bean.androidId + bean.sessionId))

    // 将结果输出位parquet文件
    logBeanRDD.coalesce(1).toDF().write.parquet(outpath)


    // 将结果输出为ORC文件 -- ORC的压缩性能、查询性能，都略强于parquet
    //logBeanRDD.coalesce(1).toDF().write.orc("G:/sharkdata/geotest/orcout/")

    val end = System.currentTimeMillis()

    System.err.println("总耗时(s)： " + (end - start) / 1000)

    spark.close()

  }

}
